{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "并发编程有两种基本模型，一种是message passing，另一种是shared memory。\n",
    "\n",
    "- 在分布式系统中，运行在多台机器上的多个进程的并行编程只有一种实用模型 message passing。\n",
    "- 在单机上，我们也可以照搬 message passing 作为多个进程的并发模型。这样整个分布式系统的架构的一致性很强，扩容（scale out）起来也较容易。\n",
    "- 在多线程编程中，message passing 更容易保证程序的正确性，有的语言只提供这一种模型。\n",
    "\n",
    "不过在用C/C++编写多线程程序时，我们仍然需要了解底层的 shared memory 模型下的同步原语，以备不时之需。\n",
    "\n",
    "本章不是[多线程教程](https://computing.llnl.gov/tutorials/pthreads)，而是个人经验总结，分享一些C++多线程编程的经验。本章多次引用《Real-World Concurrency》一文的观点，这篇文章的地址是 http://queueacm.org/detailcfm?id=1454462，后文简称[RW]。\n",
    "\n",
    "> - [https://computing.llnl.gov/tutorials/pthreads/](https://computing.llnl.gov/tutorials/pthreads/) \n",
    "> - [https://hpc-tutorials.llnl.gov/posix/](https://hpc-tutorials.llnl.gov/posix/)\n",
    "\n",
    "\n",
    "线程同步的四项原则，按重要性排列∶\n",
    "1. 首要原则是尽量最低限度地共享对象，减少需要同步的场合。一个对象能不暴露给别的线程就不要暴露; 如果要暴露，优先考虑 immutable 对象; 实在不行才暴露可修改的对象，并用同步措施来充分保护它。\n",
    "2. 其次是使用高级的并发编程构件，如 TaskQueue、Producer-Consumer Queue、CountDownLatch(倒计时门闩)等等。\n",
    "3. 最后不得已必须使用底层同步原语（primitives）时，只用非递归的互斥器和条件变量，慎用读写锁，不要用信号量。\n",
    "4. 除了使用 atomic 整数之外，不自己编写lock-free 代码，也不要用\"内核级\"同步原语。不凭空猜测\"哪种做法性能会更好\"，比如 spin lock vs. mutex。\n",
    "> - https://docs.kernel.org/process/volatile-considered-harmful.html\n",
    "> - http://www.thinkingparallel.com/2007/02/19/please-dont-rely-on-memory-barriers-for-synchronization/\n",
    "\n",
    "前面两条很容易理解，这里着重讲一下第3条∶底层同步原语的使用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.1 互斥器（mutex）\n",
    "\n",
    "互斥器（mutex）恐怕是使用得最多的同步原语，粗略地说，它保护了临界区，任何一个时刻最多只能有一个线程在此 mutex 划出的临界区内活动。单独使用mutex 时，我们主要为了保护共享数据。\n",
    "> 请注意，本书谈的是 `Pthreads` 里的`mutex`，不是Windows里的重量级跨进程 Mutex 内核对象。\n",
    "\n",
    "我个人的原则是∶\n",
    "- 用RAII手法封装mutex 的创建、销毁、加锁、解锁这四个操作。用RAII封装这几个操作是通行的做法，这几乎是C++的标准实践，后面我会给出具体的代码示例，相信大家都已经写过或用过类似的代码了。Java 里的synchronized 语句和C#的 using 语句也有类似的效果，即*保证锁的生效期间等于一个作用域（scope ），不会因异常而忘记解锁*。\n",
    "- 只用非递归的 mutex（即不可重入的 mutex）。\n",
    "- 不手工调用`lock()`和 `unlock()`函数，一切交给栈上的 Guard 对象的构造和析构函数负责。Guard 对象的生命期正好等于临界区（分析对象在什么时候析构是C++程序员的基本功）。这样我们保证始终在同一个函数同一个scope里对某个mutex加锁和解锁。避免在 `foo()`里加锁，然后跑到 `bar()`里解锁; 也避免在不同的语句分支中分别加锁、解锁。这种做法被称为Scoped Locking。\n",
    "- 在每次构造Guard 对象的时候，思考一路上（调用栈上）已经持有的锁，防止因加锁顺序不同而导致死锁（deadlock）。由于Guard 对象是栈上对象，看函数调用栈就能分析用锁的情况，非常便利。\n",
    "\n",
    "次要原则有∶\n",
    "- 不使用跨进程的 mutex，进程间通信只用TCP sockets。\n",
    "- 加锁、解锁在同一个线程，线程 a 不能去 unlock 线程 b 已经锁住的 mutex（RAII自动保证）。\n",
    "- 别忘了解锁（RAII自动保证）。\n",
    "- 不重复解锁（RAII自动保证）。\n",
    "- 必要的时候可以考虑用`PTHREAD_MUTEX_ERRORCHECK`来排错。\n",
    "\n",
    "mutex恐怕是最简单的同步原语，按照上面的几条原则，几乎不可能用错。我自己从来没有违背过这些原则，编码时出现问题都很快能定位并修复。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 只使用非递归的 mutex\n",
    "\n",
    "谈谈我坚持使用非递归的互斥器的个人想法。\n",
    "\n",
    "Mutex分为递归（recursive）和非递归（non-recursive）两种，这是 POSIX的叫法，另外的名字是*可重人（reentrant）与非可重入*。\n",
    "\n",
    "这两种 mutex作为线程间（inter-thread）的同步工具时没有区别，它们的唯一区别在于∶同一个线程可以重复对 `recursive mutex` 加锁，但是不能重复对 `non-recursive mutex`加锁。\n",
    "\n",
    "首选非递归 mutex，绝对不是为了性能，而是为了体现设计意图。non-recursive 和 recursive 的性能差别其实不大，因为少用一个计数器，前者略快一点点而已。在同一个线程里多次对 non-recursive mutex 加锁会立刻导致死锁，我认为这是它的优点，能帮助我们思考代码对锁的期求，并且及早（在编码阶段）发现问题。\n",
    "\n",
    "毫无疑问 recursive mutex 使用起来要方便一些，因为不用考虑一个线程会自已把自己给锁死了，我猜这也是 Java 和 Windows 默认提供 recursive mutex 的原因。（Java语言自带的 intrinsic lock是可重入的，它的 util.concurrent库里提供ReentrantLock，Windows 的 CRITICAL_SECTION 也是可重入的。似乎它们都不提供轻量级的non-recursive mutex。）\n",
    "\n",
    "正因为它方便，recursive mutex 可能会隐藏代码里的一些问题。典型情况是你以为拿到一个锁就能修改对象了，没想到外层代码已经拿到了锁，正在修改（或读取）同一个对象呢。来看一个具体的例子\n",
    "\n",
    "- [NonRecursiveMutex_test.cc](../../recipes/thread/test/NonRecursiveMutex_test.cc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "class Foo\n",
    "{\n",
    " public:\n",
    "  void doit() const;\n",
    "};\n",
    "\n",
    "MutexLock mutex;\n",
    "std::vector<Foo> foos;\n",
    "\n",
    "void post(const Foo& f)\n",
    "{\n",
    "  MutexLockGuard lock(mutex);\n",
    "  foos.push_back(f);\n",
    "}\n",
    "\n",
    "void traverse()\n",
    "{\n",
    "  MutexLockGuard lock(mutex);\n",
    "  for (std::vector<Foo>::const_iterator it = foos.begin();\n",
    "      it != foos.end(); ++it)\n",
    "  {\n",
    "    it->doit();\n",
    "  }\n",
    "}\n",
    "\n",
    "void Foo::doit() const\n",
    "{\n",
    "  Foo f;\n",
    "  post(f);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`post()`加锁，然后修改`foos` 对象; `traverse()`加锁，然后遍历 `foos` 向量。这些都是正确的。\n",
    "\n",
    "将来有一天，`Foo::doit()`间接调用了`post()`，那么会很有戏剧性的结果∶\n",
    "- Mutex是非递归的，于是死锁了。\n",
    "- Mutex是递归的，由于`push_back()`可能（但不总是）导致 `vector`迭代器失效，程序偶尔会crash。\n",
    "\n",
    "这时候就能体现 non-recursive 的优越性∶把程序的逻辑错误暴露出来。死锁比较容易 debug，把各个线程的调用栈打出来，只要每个函数不是特别长，很容易看出来是怎么死的，见§2.1.2 的例子。或者可以用`PTHREAD_MUTEX_ERRORCHECK`一下子就能找到错误（前提是MutexLock 带 debug 选项）。程序反正要死，不如死得有意义一点，留个\"全尸\"，让验尸（post-mortem）更容易些。\n",
    "\n",
    "> - gdb中使用 `thread apply all bt`命令。\n",
    "> - 另一方面支持了\"函数不要写得过长\"这一观点。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "**问题概览**\n",
    "\n",
    "这段代码存在至少两个严重缺陷：\n",
    "\n",
    "1. **死锁风险（非递归互斥量）**\n",
    "   - `traverse()` 在持有 `mutex` 的情况下遍历 `foos`。\n",
    "   - 迭代过程中调用 `it->doit()`。\n",
    "   - `Foo::doit()` 再调用 `post()`，而 `post()` 会尝试给同一个 `mutex` 上锁。\n",
    "   - 如果 `MutexLock` 是通常的“非递归”互斥量（推荐类型），同一线程第二次 `lock` 会直接导致死锁。\n",
    "   - 这正好违背了我们在前面讨论的同步原则：“只用非递归互斥器，并避免递归加锁”。\n",
    "2. **迭代器失效风险（即使是递归互斥量）**\n",
    "   - 假设 `MutexLock` 恰好是递归锁，`post()` 可以成功进入，但 `foos.push_back(f)` 可能触发扩容。\n",
    "   - 扩容会重新分配底层缓冲区，使得 `traverse()` 持有的所有迭代器和引用立即失效，继续访问 `*it` 会产生未定义行为。\n",
    "   - 因此即便“巧合”地没有死锁，也会在重新分配时崩溃或读到垃圾数据。\n",
    "\n",
    "**总结**\n",
    "\n",
    "- **不要在持锁的情况下调用可能再次触发加锁或修改共享容器的回调。**\n",
    "- 如果确实需要让 `Foo::doit()` 追加元素，可以采取：\n",
    "  - 在 `traverse()` 中先复制 `foos` 的内容或提取需要处理的指针，在释放锁之后再执行回调；\n",
    "  - 或者让 `doit()` 将任务排入其他线程／队列，而不是直接调用需要锁的函数；\n",
    "  - 尽量避免边遍历边修改同一个 `std::vector`。\n",
    "\n",
    "这段代码的核心问题是“在锁保护区里回调用户代码”，导致同步与容器安全两方面都出错。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void traverse()\n",
    "{\n",
    "  std::vector<Foo> pending;\n",
    "\n",
    "  {\n",
    "    // 把要处理的元素一次性搬到局部变量，释放互斥后再执行回调\n",
    "    MutexLockGuard lock(mutex);\n",
    "    pending.swap(foos); // swap 避免了逐个复制，同时保证迭代器不会在调用期间失效\n",
    "  } // traverse() 先在持锁区内把 foos 的内容搬到局部 pending 中，然后立即释放互斥量\n",
    "\n",
    "  for (std::vector<Foo>::const_iterator it = pending.begin();\n",
    "       it != pending.end(); ++it)\n",
    "  {\n",
    "    it->doit(); // 回调 doit() 在锁外执行，即使它内部调用 post() 也不会造成递归加锁。\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果确实需要在遍历的时候修改 `vector`，有两种做法，\n",
    "- 一是把修改推后，记住循环中试图添加或删除哪些元素，等循环结束了再依记录修改 `foos`; \n",
    "- 二是用copy-on-write，见 §2.8的例子。\n",
    "\n",
    "如果一个函数既可能在已加锁的情况下调用，又可能在未加锁的情况下调用，那么就拆成两个函数∶\n",
    "- 跟原来的函数同名，函数加锁，转而调用第2个函数。\n",
    "- 给函数名加上后缀`WithLockHold`，不加锁，把原来的函数体搬过来。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void post(const Foo& f)\n",
    "{\n",
    "    MutexLockGuard lock(mutex); \n",
    "    postWithLockHold(f); // 不用担心开销，编译器会自动内联的\n",
    "}\n",
    "\n",
    "// 引入这个函数是为了体现代码作者的意图，尽管push_back通常可以手动内联\n",
    "void postWithLockHold(const Foo& f)\n",
    "{\n",
    "    foos.push_back(f);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这有可能出现两个问题（感谢水木网友 ilovecpp提出）∶\n",
    "\n",
    "1. 误用了加锁版本，死锁了。\n",
    "2. 误用了不加锁版本，数据损坏了。\n",
    "\n",
    "对于（1），仿造§2.1.2的办法能比较容易地排错。对于（2），如果Pthreads提供`isLockedByThisThread()`就好办，可以写成∶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void postWithLockHold(const Foo& f)\n",
    "{\n",
    "    assert(mutex.isLockedByThisThread());\n",
    "    // ... \n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "另外，WithLockHold 这个显眼的后缀也让程序中的误用容易暴露出来。\n",
    "\n",
    "C++没有annotation，不能像 Java那样给 method 或 field标上`@GuardedBy` 注解，需要程序员自己小心在意。虽然这里的办法不能一劳永逸地解决全部多线程错误，但能帮上一点是一点了。\n",
    "\n",
    "我还没有遇到过需要使用 recursive mutex 的情况，我想将来遇到了都可以借助wrapper 改用non-recursive mutex，代码只会更清晰。\n",
    "> Pthreads 的权威专家，《Programming with POSIX Threads》的作者 DavidButenhof 也[排斥使用 recursive mutex](http://zaval.org/resources/library/butenhof1.html)。\n",
    "\n",
    "回到正题。本文这里只谈了mutex本身的正确使用，在 C++里多线程编程还会遇到其他一些 race condition，请参看第1章。\n",
    "\n",
    "性能注脚∶Linux的Pthreads mutex采用 futex（2）[实现](http://www.akkadia.org/drepper/futex.pdf)，不必每次加锁、解锁都陷入系统调用，效率不错。Windows 的 [CRITICAL_SECTION](https://learn.microsoft.com/en-us/windows/win32/sync/critical-section-objects?redirectedfrom=MSDN) 也是类似的，不过它可以嵌入一小段spin lock。在多CPU系统上，如果不能立刻拿到锁，它会先spin一小段时间，如果还不能拿到锁，才挂起当前线程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## * 死锁\n",
    "\n",
    "前面说过，如果坚持只使用 Scoped Locking，那么在出现死锁的时候很容易定位。考虑下面这个线程自己与自己死锁的例子\n",
    "\n",
    "- [SelfDeadLock.cc](../../recipes/thread/test/SelfDeadLock.cc)\n",
    "\n",
    "在上面这个例子中，原本没有 L9，在添加它之后，程序立刻出现了死锁。要调试定位这种死锁很容易，只要把函数调用栈打印出来，结合源码一看，我们立刻就会发现第6帧 `Request::process()`和第5帧 `Request::print()`先后对同一个mutex上锁，引发了死锁。（必要的时候可以加上`__attribute__`来防止函数 `inline` 展开。）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include <mutex>\n",
    "#include <iostream>\n",
    "\n",
    "class Request\n",
    "{\n",
    "public:\n",
    "  void process()\n",
    "  {\n",
    "    std::lock_guard<std::mutex> lock(mutex_);\n",
    "    print();  // 二次加锁：复现死锁\n",
    "    std::cout << \"process finished\\n\";  // 永远不会到达\n",
    "  }\n",
    "\n",
    "  void print() const\n",
    "  {\n",
    "    std::lock_guard<std::mutex> lock(mutex_);  // 与 process() 使用同一把非递归 mutex\n",
    "    std::cout << \"print called\\n\";\n",
    "  }\n",
    "\n",
    "private:\n",
    "  mutable std::mutex mutex_;\n",
    "};\n",
    "\n",
    "int main()\n",
    "{\n",
    "  Request req;\n",
    "  req.process();  // 程序在此处死锁\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "(gdb) thread apply all bt\n",
    "\n",
    "Thread 1 (Thread 0x7ffff7a02740 (LWP 3408) \"a.out\"):\n",
    "#0  futex_wait (private=0, expected=2, futex_word=0x7fffffffead0) at ../sysdeps/nptl/futex-internal.h:146\n",
    "#1  __GI___lll_lock_wait (futex=futex@entry=0x7fffffffead0, private=0) at ./nptl/lowlevellock.c:49\n",
    "#2  0x00007ffff7b900f1 in lll_mutex_lock_optimized (mutex=0x7fffffffead0) at ./nptl/pthread_mutex_lock.c:48\n",
    "#3  ___pthread_mutex_lock (mutex=0x7fffffffead0) at ./nptl/pthread_mutex_lock.c:93\n",
    "#4  0x000055555555526d in __gthread_mutex_lock (__mutex=0x7fffffffead0) at /usr/include/x86_64-linux-gnu/c++/14/bits/gthr-default.h:762\n",
    "#5  0x00005555555552c2 in std::mutex::lock (this=0x7fffffffead0) at /usr/include/c++/14/bits/std_mutex.h:113\n",
    "#6  0x000055555555547e in std::lock_guard<std::mutex>::lock_guard (this=0x7fffffffea60, __m=...) at /usr/include/c++/14/bits/std_mutex.h:250\n",
    "#7  0x00005555555553dd in Request::print (this=0x7fffffffead0) at main.cpp:16\n",
    "#8  0x0000555555555337 in Request::process (this=0x7fffffffead0) at main.cpp:10\n",
    "#9  0x000055555555522c in main () at main.cpp:27"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "从这份 backtrace 可以看出，唯一的线程（`Thread 1`）正卡在同一把互斥量上，形成了**自死锁**：\n",
    "\n",
    "1. 栈顶在 `futex_wait / __pthread_mutex_lock`（#0–#4），说明线程正阻塞在 `std::mutex::lock()` 上（#5）。\n",
    "2. 调用链显示这个 `lock()` 来自 `Request::print()` 的 `std::lock_guard<std::mutex>` 构造函数（#6–#7）。\n",
    "3. 继续往下看，`print()` 是由 `Request::process()` 调用的（#8），而 `process()` 在调用 `print()` 前已经构造了一个 `std::lock_guard<std::mutex>`（因此同一线程持有这把锁）。\n",
    "4. 由于 `std::mutex` 不是递归锁，线程在 `process()` 中拿到锁后、再次在 `print()` 中尝试加锁，就只能阻塞等待自己释放——于是形成死锁。\n",
    "5. `thread apply all bt` 只显示一个线程，这也进一步证明没有其他线程来解锁，它完全是“锁住自己”。\n",
    "\n",
    "因此，单线程在 `process()`→`print()` 的链条上对同一把非递归互斥量进行两次加锁，直接导致了死锁。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "要修复这个错误也很容易，按前面的办法，从 `Request::print()`抽取出`Request::printWithLockHold()`，并让 `Request::print()`和`Request::process()`都调用它即可。\n",
    "\n",
    "> 即extract method重构手法。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include <iostream>\n",
    "#include <mutex>\n",
    "\n",
    "class Request\n",
    "{\n",
    "public:\n",
    "  void process()\n",
    "  {\n",
    "    std::lock_guard<std::mutex> lock(mutex_);\n",
    "    printWithLockHold();\n",
    "  }\n",
    "\n",
    "  void print() const\n",
    "  {\n",
    "    std::lock_guard<std::mutex> lock(mutex_);\n",
    "    printWithLockHold();\n",
    "  }\n",
    "\n",
    "private:\n",
    "  void printWithLockHold() const\n",
    "  {\n",
    "    std::cout << \"Request::print()\" << std::endl;\n",
    "  }\n",
    "\n",
    "  mutable std::mutex mutex_;\n",
    "};"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "再来看一个更真实的两个线程死锁的例子\n",
    "\n",
    "有一个`Inventory class`（清单），记录当前的 `Request`对象。容易看出，下面这个`Inventory class`的`add()`和`remove()`成员函数都是线程安全的，它使用了mutex来保护共享数据 `requests_`。\n",
    "\n",
    "> 为了简单起见，这里没有使用第1章介绍的`shared_ptr/weak_ptr`来管理Request。\n",
    "\n",
    "- [MutualDeadLock.cc](../../recipes/thread/test/MutualDeadLock.cc#L8)\n",
    "\n",
    "`Request class`与`Inventory class`的交互逻辑很简单，在处理（`process`）请求的时候，往`g_inventory`中添加自己。在析构的时候，从`g_inventory`中移除自己。目前看来，整个程序还是线程安全的。\n",
    "\n",
    "- [MutualDeadLock.cc](../../recipes/thread/test/MutualDeadLock.cc#L32)\n",
    "\n",
    "`Inventory class` 还有一个功能是打印全部已知的` Request` 对象。`Inventory::printAll()`里的逻辑单独看是没问题的，但是它有可能引发死锁。\n",
    "\n",
    "- [MutualDeadLock.cc](../../recipes/thread/test/MutualDeadLock.cc#L59)\n",
    "\n",
    "下面程序运行起来发生了死锁\n",
    "\n",
    "- [MutualDeadLock.cc](../../recipes/thread/test/MutualDeadLock.cc#L89)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "! g++ -std=c++17 -pthread -g deadlock_demo.cpp -o deadlock_demo"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "通过gdb查看两个线程的函数调用栈, 我们发现两个线程都等在mutex上(`__lll__lock__wait`), 估计是发生了死锁. 因为一个程序中的线程一般只会等在condition variable上, 或者等在epoll_wait上"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "shellscript"
    }
   },
   "outputs": [],
   "source": [
    "(gdb) thread apply all bt\n",
    "\n",
    "Thread 2 (Thread 0x7ffff77ff640 (LWP 284222) \"deadlock_demo\"):\n",
    "#0  futex_wait (private=0, expected=2, futex_word=0x55555555d040 <g_inventory>) at ../sysdeps/nptl/futex-internal.h:146\n",
    "#1  __GI___lll_lock_wait (futex=futex@entry=0x55555555d040 <g_inventory>, private=0) at ./nptl/lowlevellock.c:49\n",
    "#2  0x00007ffff7898002 in lll_mutex_lock_optimized (mutex=0x55555555d040 <g_inventory>) at ./nptl/pthread_mutex_lock.c:48\n",
    "#3  ___pthread_mutex_lock (mutex=0x55555555d040 <g_inventory>) at ./nptl/pthread_mutex_lock.c:93\n",
    "#4  0x0000555555555779 in __gthread_mutex_lock (__mutex=0x55555555d040 <g_inventory>) at /usr/include/x86_64-linux-gnu/c++/11/bits/gthr-default.h:749\n",
    "#5  0x00005555555557ce in std::mutex::lock (this=0x55555555d040 <g_inventory>) at /usr/include/c++/11/bits/std_mutex.h:100\n",
    "#6  0x0000555555555cca in std::lock_guard<std::mutex>::lock_guard (this=0x7ffff77fed00, __m=...) at /usr/include/c++/11/bits/std_mutex.h:229\n",
    "#7  0x0000555555555989 in Inventory::remove (this=0x55555555d040 <g_inventory>, req=0x7ffff0000b70) at deadlock_demo.cpp:20\n",
    "#8  0x0000555555555bee in Request::~Request (this=0x7ffff0000b70, __in_chrg=<optimized out>) at deadlock_demo.cpp:47\n",
    "#9  0x000055555555552b in threadFunc () at deadlock_demo.cpp:94\n",
    "#10 0x00005555555582ab in std::__invoke_impl<void, void (*)()> (__f=@0x55555556feb8: 0x5555555554e8 <threadFunc()>) at /usr/include/c++/11/bits/invoke.h:61\n",
    "#11 0x0000555555558257 in std::__invoke<void (*)()> (__fn=@0x55555556feb8: 0x5555555554e8 <threadFunc()>) at /usr/include/c++/11/bits/invoke.h:96\n",
    "#12 0x00005555555581f8 in std::thread::_Invoker<std::tuple<void (*)()> >::_M_invoke<0ul> (this=0x55555556feb8) at /usr/include/c++/11/bits/std_thread.h:259\n",
    "#13 0x00005555555581c8 in std::thread::_Invoker<std::tuple<void (*)()> >::operator() (this=0x55555556feb8) at /usr/include/c++/11/bits/std_thread.h:266\n",
    "#14 0x00005555555581a8 in std::thread::_State_impl<std::thread::_Invoker<std::tuple<void (*)()> > >::_M_run (this=0x55555556feb0) at /usr/include/c++/11/bits/std_thread.h:211\n",
    "#15 0x00007ffff7cdc253 in ?? () from /lib/x86_64-linux-gnu/libstdc++.so.6\n",
    "#16 0x00007ffff7894ac3 in start_thread (arg=<optimized out>) at ./nptl/pthread_create.c:442\n",
    "#17 0x00007ffff79268c0 in clone3 () at ../sysdeps/unix/sysv/linux/x86_64/clone3.S:81\n",
    "\n",
    "Thread 1 (Thread 0x7ffff7ea73c0 (LWP 284219) \"deadlock_demo\"):\n",
    "#0  futex_wait (private=0, expected=2, futex_word=0x7ffff0000b70) at ../sysdeps/nptl/futex-internal.h:146\n",
    "#1  __GI___lll_lock_wait (futex=futex@entry=0x7ffff0000b70, private=0) at ./nptl/lowlevellock.c:49\n",
    "#2  0x00007ffff7898002 in lll_mutex_lock_optimized (mutex=0x7ffff0000b70) at ./nptl/pthread_mutex_lock.c:48\n",
    "#3  ___pthread_mutex_lock (mutex=0x7ffff0000b70) at ./nptl/pthread_mutex_lock.c:93\n",
    "#4  0x0000555555555779 in __gthread_mutex_lock (__mutex=0x7ffff0000b70) at /usr/include/x86_64-linux-gnu/c++/11/bits/gthr-default.h:749\n",
    "#5  0x00005555555557ce in std::mutex::lock (this=0x7ffff0000b70) at /usr/include/c++/11/bits/std_mutex.h:100\n",
    "#6  0x0000555555555cca in std::lock_guard<std::mutex>::lock_guard (this=0x7fffffffd810, __m=...) at /usr/include/c++/11/bits/std_mutex.h:229\n",
    "#7  0x0000555555555c44 in Request::print (this=0x7ffff0000b70) at deadlock_demo.cpp:52\n",
    "#8  0x000055555555545c in Inventory::printAll (this=0x55555555d040 <g_inventory>) at deadlock_demo.cpp:68\n",
    "#9  0x00005555555555a6 in main () at deadlock_demo.cpp:101"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这两个调用序列对两个mutex 的加锁顺序正好相反，于是造成了经典的死锁。见图 2-1，`Inventory class`的mutex的临界区由灰底表示，`Request class`的mutex的临界区由斜纹表示。一旦`main()`线程中的 `printAll()`在另一个线程的`~Request()`和`remove()`之间开始执行，死锁已不可避免。\n",
    "\n",
    "<img src=\"./images/2.1.png\" alt=\"2.1\" style=\"zoom:50%;\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- **主线程**：持有 `Inventory::mutex_`，在 `Inventory::printAll()` 调用 `(*it)->print()` 时尝试获取 `Request::mutex_`，但被另一线程占着。\n",
    "- **子线程**：持有 `Request::mutex_`（在 `Request::~Request()` 中），等待 `Inventory::mutex_`（准备调用 `g_inventory.remove(this)`），也被主线程占住。\n",
    "\n",
    "两者互相等待形成环路：\n",
    "\n",
    "- 主线程（printAll）：已锁 `Inventory` → 等待 `Request`\n",
    "- 子线程（~Request）：已锁 `Request` → 等待 `Inventory`\n",
    "   正是典型的锁顺序反转死锁。\n",
    "\n",
    "**为何能稳定复现？**\n",
    "\n",
    "- `main()` 里先启动 `threadFunc`，让 `Request` 对象调用 `process()` 把自己加入 `g_inventory`，然后主线程 `sleep_for(500ms)` 给子线程时间准备。\n",
    "- 此时主线程进入 `g_inventory.printAll()`，持有 `Inventory::mutex_` 后 `sleep(1s)`，为死锁制造时机。\n",
    "- 子线程删除对象时进入 `~Request()`，先持有 `Request::mutex_`，`sleep(1s)`（确保主线程还在拿着 `Inventory` 锁），接着想 `remove` 自己，会尝试获取 `Inventory::mutex_`，但被主线程占着 → 阻塞。\n",
    "- 与此同时主线程在 `printAll()` 的循环里，调用 `(*it)->print()` 想获取 `Request::mutex_`，发现 `Request` 锁被子线程持有 → 再阻塞。\n",
    "- 两边都在等待对方释放锁，死锁形成。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "思考∶如果`printAll`晚于`remove()`执行，还会出现死锁吗?\n",
    "\n",
    "练习∶修改程序，让`~Request()`在 `printAll()`和`print()`之间开始执行，复现另一种可能的死锁时序。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这里也出现了第1章所说的对象析构的 race condition，即一个线程正在析构对象，另一个线程却在调用它的成员函数。\n",
    "\n",
    "解决死锁的办法很简单，要么把`print()`移出 `printAll()`的临界区，这可以用§2.8介绍的办法; 要么把 `remove()`移出`~Request()`的临界区，比如交换p.37中`muduo::MutexLockGuard lock(mutex_);`和 `g_inventory.remove(this);` 两行代码的位置。当然这没有解决对象析构的 race condition，留给读者当做练习吧。\n",
    "\n",
    "思考∶`Inventory::printAll`→ `Request::print`有没有可能与`Request::process`→`Inventory::add`发生死锁?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "死锁会让程序行为失常，其他一些锁使用不当则会影响性能，例如潘爱民老师写的《Lock Convoys Explained》详细解释了一种性能衰退的现象。除此之外，编写高性能多线程程序至少还要知道 false sharing 和 CPU cache效应，可看这几篇入门\n",
    "\n",
    "- [https://nwcpp.org/talks/2007/Machine_Architecture_-_NWCPP.pdf](https://nwcpp.org/talks/2007/Machine_Architecture_-_NWCPP.pdf)\n",
    "- [https://www.aristeia.com/TalkNotes/ACCU2011_CPUCaches.pdf](https://www.aristeia.com/TalkNotes/ACCU2011_CPUCaches.pdf)\n",
    "- [http://igoro.com/archive/gallery-of-processor-cache-effects/](http://igoro.com/archive/gallery-of-processor-cache-effects/)\n",
    "- [http://simplygenius.net/Article/FalseSharing](http://simplygenius.net/Article/FalseSharing)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.2 条件变量（condition variable）\n",
    "\n",
    "互斥器（mutex）是加锁原语，用来排他性地访问共享数据，它不是等待原语。在使用 mutex 的时候，我们一般都会期望加锁不要阻塞，总是能立刻拿到锁。然后尽快访问数据，用完之后尽快解锁，这样才能不影响并发性和性能。\n",
    "\n",
    "如果需要等待某个条件成立，我们应该使用条件变量（condition variable）。条件变量顾名思义是一个或多个线程等待某个布尔表达式为真，即等待别的线程\"唤醒\"它。条件变量的学名叫管程（monitor）。\n",
    "\n",
    "Java Object内置的 `wait()`、`notify()`、`notifyAll()`是条件变量\n",
    "> Java 的这三个函数以容易用错著称，一般建议用`java.util.concurrent`中的同步原语。\n",
    "\n",
    "条件变量只有一种正确使用的方式，几乎不可能用错。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于wait 端∶\n",
    "- 必须与mutex一起使用，该布尔表达式的读写需受此mutex保护。\n",
    "- 在mutex已上锁的时候才能调用`wait()`。\n",
    "- 把判断布尔条件和`wait()`放到 `while`循环中。\n",
    "\n",
    "写成代码是∶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "muduo::MutexLock  mutex;\n",
    "muduo::Condition  cond(mutex);\n",
    "std::deque<int>   queue;\n",
    "\n",
    "int dequeue()\n",
    "{\n",
    "  MutexLockGuard lock(mutex);\n",
    "  while (queue.empty())          // 必须用循环; 必须在判断之后再wait()\n",
    "  {\n",
    "    cond.wait();                 // 这一步会原子地unlock mutex并进入等待，不会与enqueue死锁 \n",
    "\n",
    "    // wait() 执行完毕时会自动重新加锁\n",
    "  }\n",
    "\n",
    "  assert(!queue.empty());\n",
    "  int top = queue.front();\n",
    "  queue.pop_front();\n",
    "\n",
    "  return top;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "上面的代码中必须用`while`循环来等待条件变量，而不能用`if`语句，原因是spurious wakeup。这也是面试多线程编程的常见考点。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "在实现线程同步时，必须使用`while`循环而不是`if`语句来等待条件变量的原因是防止虚假唤醒（spurious wakeup），而不是死锁。虚假唤醒是指在没有明确通知时，等待线程会偶尔（虽然很罕见）从等待状态中唤醒。在这种情况下，如果我们使用`if`语句来等待条件变量，那么程序将错误地假定条件已满足并继续执行，在这种情况下可以使用`while`循环重新等待直到条件满足。\n",
    "\n",
    "使用`while`循环可以确保虚假唤醒不会导致程序不正确的行为。被唤醒的线程会重新检查条件是否满足，如果条件不满足，那么该线程将继续等待条件变量。如果我们使用`if`语句，则可能在条件未满足的情况下返回，这可能会导致一些非预期的情况，例如死锁或竞争条件等。\n",
    "\n",
    "在上面的代码中，使用了`while`循环来等待条件变量。当队列为空时，等待线程将进入等待状态，直到其他线程通过唤醒等待线程的方式向队列中添加新元素并通知等待线程。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "对于 signal/broadcast端∶\n",
    "\n",
    "- 不一定要在 mutex 已上锁的情况下调用 signal（理论上）。\n",
    "- 在 signal 之前一般要修改布尔表达式。\n",
    "- 修改布尔表达式通常要用mutex保护（至少用作 full memory barrier）。\n",
    "- 注意区分signal与broadcast∶ \n",
    "  - broadcast 通常用于表明状态变化，\n",
    "  - signal 通常用于表示资源可用。\n",
    "  - broadcast should generally be used to indicate state change rather than resource availability\n",
    "\n",
    "写成代码\n",
    "\n",
    "> `muduo::Condition`采用了`notify()`和 `notifyAll()`为函数名，避免重载signal这个术语。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void enqueue(int x)\n",
    "{\n",
    "    MutexLockGuard lock(mutex);\n",
    "    queue.push_back(x); \n",
    "    cond.notify(); // 可以移出临界区之外\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "上面的`dequeue() / enqueue()`实际上实现了一个简单的容量无限的(unbounded) BlockingQueue\n",
    "> 实际使用时一般会做成类模板，如[muduo/base/BlockingQueue.h](../../muduo/muduo/base/BlockingQueue.h)\n",
    "\n",
    "思考∶`enqueue()`中每次添加元素都会调用`Condition::notify()`，如果改成只在`queue.size()`从0变1的时候才调用`Condition::notify()`，[会出现什么后果](https://blog.csdn.net/Solstice/article/details/5829421#comments)?\n",
    "\n",
    "条件变量是非常底层的同步原语，很少直接使用，一般都是用它来实现高层的同步措施，如 `BlockingQueue<T>` 或 `CountDownLatch`。\n",
    "\n",
    "倒计时（CountDownLatch）是一种常用且易用的同步手段。\n",
    "\n",
    "它主要有两种用途:\n",
    "- 主线程发起多个子线程，等这些子线程各自都完成一定的任务之后，主线程才继续执行。通常用于主线程等待多个子线程完成初始化。\n",
    "- 主线程发起多个子线程，子线程都等待主线程，主线程完成其他一些任务之后通知所有子线程开始执行。通常用于多个子线程等待主线程发出\"起跑\"命令。\n",
    "\n",
    "当然我们可以直接用条件变量来实现以上两种同步。不过如果用`CountDownLatch`的话，程序的逻辑更清晰。\n",
    "\n",
    "`CountDownLatch` 的接口很简单∶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "class CountDownLatch : boost::noncopyable\n",
    "{\n",
    "public:\n",
    "\tCountDownLatch(int count) :mutex_(), \n",
    "                            condition_(mutex_), // 初始化顺序要与成员声明保持一致\n",
    "                            count_(count){}\n",
    "\n",
    "\texplicit CounterLatch(int count); // 倒数几次\n",
    "\tvoid wait();                      // 等待计数值变为0\n",
    "    void countDown();                 // 计数减一\n",
    "private:\n",
    "\tmutable MutexLock mutex_; // 顺序很重要, 先mutex后condition\n",
    "    Condition condition_;\n",
    "    int count_;\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`CountDownLatch` 的实现同样简单，几乎就是条件变量的教科书式应用∶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void CountDownLatch::wait()\n",
    "{\n",
    "    MutexLockGuard lock(mutex_); \n",
    "    while(count_> 0) \n",
    "        condition_.wait();\n",
    "}\n",
    "\n",
    "void CountDownLatch::countDown()\n",
    "{\n",
    "    MutexLockGuard lock(mutex_); \n",
    "    --count_;\n",
    "    if(count_== 0)\n",
    "        condition_.notifyAll();\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "注意到`CountDownLatch::countDown()`使用的是`Condition::notifyAll()`，而前面p.41的`enqueue()`使用的是`Condition::notify()`，这都是有意为之。请读者思考，如果交换两种用法会出现什么情况?\n",
    "\n",
    "互斥器和条件变量构成了多线程编程的全部必备同步原语，用它们即可完成任何多线程同步任务，二者不能相互替代。我认为应该精通这两个同步原语的用法，先学会编写正确的、安全的多线程程序，再在必要的时候考虑用其他\"高技术\"手段提高性能，如果确实能提高性能的话。千万不要连 mutex都还没学会、用好，一上来就考虑lock-free设计。\n",
    "\n",
    "> - 就像与非门和D触发器构成了数字电路设计所需的全部基础元件一样，用它们可以完成任何组合和同步时序逻辑电路设计。\n",
    "> - https://jacobfilipp.com/DrDobbs/articles/DDJ/2008/0809/080801hs01/080801hs01.html"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> C++20实现了`CountDownLatch`/倒计数门闩, 即`std::latch`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "#include <latch>\n",
    "#include <thread>\n",
    "#include <vector>\n",
    "#include <iostream>\n",
    "\n",
    "int main() {\n",
    "    std::latch latch{3};\n",
    "\n",
    "    auto worker = [&latch](int id) {\n",
    "        // ... 执行任务 ...\n",
    "        latch.count_down();   // 完成后递减\n",
    "    };\n",
    "\n",
    "    std::vector<std::thread> threads;\n",
    "    for (int i = 0; i < 3; ++i) {\n",
    "        threads.emplace_back(worker, i);\n",
    "    }\n",
    "\n",
    "    latch.wait();             // 等待计数归零\n",
    "    std::cout << \"All done\\n\";\n",
    "\n",
    "    for (auto& t : threads) {\n",
    "        t.join();\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.3 不要用读写锁和信号量\n",
    "\n",
    "读写锁（Readers-Writer lock，简写为rwlock）是个看上去很美的抽象，它明确区分了 read 和 write 两种行为。\n",
    "\n",
    "初学者常干的一件事情是，一见到某个共享数据结构频繁读而很少写，就把 mutex 替换为rwlock。甚至首选 rwlock 来保护共享状态，这不见得是正确的。\n",
    "\n",
    "- 从正确性方面来说，一种典型的易犯错误是在持有read lock的时候修改了共享数据。这通常发生在程序的维护阶段，为了新增功能，程序员不小心在原来read lock 保护的函数中调用了会修改状态的函数。这种错误的后果跟无保护并发读写共享数据是一样的。\n",
    "- 从性能方面来说，读写锁不见得比普通mutex更高效。无论如何reader lock加锁的开销不会比mutex lock小，因为它要更新当前 reader的数目。如果临界区很小，锁竞争不激烈，那么mutex往往会更快。见§1.9的例子。\n",
    "> 在多线程编程中，我们总是设法缩短临界区，不是吗?\n",
    "- reader lock可能允许提升（upgrade）为writer lock，也可能不允许提升(`Pthreads rwlock`不允许提升)。考虑§2.1.1的`post()`和 `traverse()`示例，如果用读写锁来保护 `foos` 对象，那么`post()`应该持有写锁，而 `traverse()`应该持有读锁。如果允许把读锁提升为写锁，后果跟使用recursive mutex一样，会造成迭代器失效，程序崩溃。如果不允许提升，后果跟使用non-recursive mutex一样，会造成死锁。我宁愿程序死锁，留个\"全尸\"好查验。\n",
    "- 通常reader lock 是可重入的，writer lock是不可重入的。但是为了防止 writer饥饿，writer lock 通常会阻塞后来的 reader lock，因此 reader lock在重入的时候可能死锁。另外，在追求低延迟读取的场合也不适用读写锁，见 p.55。\n",
    "\n",
    "muduo 线程库有意不提供读写锁的封装，因为我还没有在工作中遇到过用rwlock 替换普通 mutex 会显著提高性能的例子。相反，我们一般建议首选 mutex。\n",
    "\n",
    "遇到并发读写，如果条件合适，我通常会用S2.8的办法，而不用读写锁，同时避免reader 被 writer 阻塞。如果确实对并发读写有极高的性能要求，可以考虑read-copy-update。\n",
    "\n",
    "信号量（Semaphore ）∶我没有遇到过需要使用信号量的情况，无从谈及个人经验。我认为信号量不是必备的同步原语。因为条件变量配合互斥器可以完全替代其功能，而且更不易用错。除了[RWC]指出的\"semaphore has no notion of ownership\"之外，信号量的另一个问题在于它有自己的计数值，而通常我们自己的数据结构也有长度值，这就造成了同样的信息存了两份，需要时刻保持一致，这增加了程序员的负担和出错的可能。如果要控制并发度，可以考虑用`muduo::ThreadPool`。\n",
    "\n",
    "说一句不知天高地厚的话，如果程序里需要解决如\"哲学家就餐\"之类的复杂IPC问题，我认为应该首先检讨这个设计∶为什么线程之间会有如此复杂的资源争抢（一个线程要同时抢到两个资源，一个资源可以被两个线程争夺）?如果在工作中遇到，我会把\"想吃饭\"这个事情专门交给一个为各位哲学家分派餐具的线程来做，然后每个哲学家等在一个简单的 condition variable上，到时间了有人通知他去吃饭。从哲学上说，教科书上的解决方案是平权，每个哲学家有自己的线程，自己去拿筷子;我宁愿用集权的方式，用一个线程专门管餐具的分配，让其他哲学家线程拿个号等在食堂门口好了。这样不损失多少效率，却让程序简单很多。虽然Windows 的 `WaitForMultipleObjects` 让这个问题 trivial化，但在Linux下正确模拟`WaitForMultipleObjects`不是普通程序员该干的。\n",
    "\n",
    "`Pthreads` 还提供了`barrier` 这个同步原语，我认为不如 `CountDownLatch`实用。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.4 封装 MutexLock、MutexLockGuard、Condition\n",
    "\n",
    "本节把前面用到的 `MutexLock`、`MutexLockGuard`、`Condition` 等 class 的代码列出来，前面两个class没多大难度，后面那个有点意思。\n",
    "\n",
    "这几个 class都不允许拷贝构造和赋值。完整代码可以在 muduo/base 找到。\n",
    "\n",
    "- [class MutexLock](../../recipes/thread/Mutex.h#L20)\n",
    "- [class MutexLockGuard](../../recipes/thread/Mutex.h#L70)\n",
    "\n",
    "`MutexLock` 和 `MutexLockGuard` 这两个class 应该能在纸上默写出来，没有太多需要解释的。\n",
    "\n",
    "注意上面代码的最后一行定义了一个宏，这个宏的作用是防止程序里出现如下错误;"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void doit()\n",
    "{\n",
    "    MutexLockGuard(mutex); //遗漏变量名，产生一个临时对象又马上销毁了，结果没有锁住临界区。\n",
    "    // 正确写法是 MutexLockGuard lock(mutex);\n",
    "    // 临界区\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "我见过有人把 `MutexLockGuard` 写成 template，我没有这么做是因为它的模板类型参数只有`MutexLock` 一种可能，没有必要随意增加灵活性，于是我手工把模板具现化（instantiate）了\n",
    "\n",
    "此外一种更激进的写法是，把`lock/unlock`放到 `private` 区，然后把`MutexLockGuard`设为`MutexLock` 的 `friend`。我认为在注释里告知程序员即可，另外 check-in 之前的code review也很容易发现误用的情况（`grep getPthreadMutex`）。\n",
    "\n",
    "这段代码没有达到工业强度∶\n",
    "- mutex 创建为 `PTHREAD_MUTEX_DEFAULT`类型，而不是我们预想的 `PTHREAD_MUTEX_NORMAL` 类型（实际上这二者很可能是等同的），严格的做法是用 `mutexattr`来显示指定mutex 的类型。\n",
    "- 没有检查返回值。这里不能用 `assert()`检查返回值，因为 `assert()`在release build 里是空语句。我们检查返回值的意义在于防止 `ENOMEM`之类的资源不足情况，这一般只可能在负载很重的产品程序中出现。一旦出现这种错误，程序必须立刻清理现场并主动退出，否则会莫名其妙地崩溃，给事后调查造成困难。这里我们需要non-debug的 assert，或许 google-glog 的`CHECK()`宏是个不错的思路。\n",
    "\n",
    "以上两点改进留作练习。\n",
    "\n",
    "muduo 库的一个特点是只提供最常用、最基本的功能，特别有意避免提供多种功能近似的选择。muduo 不是\"杂货铺\"，不会不分青红皂白地把各种有用的、没用的功能全铺开摆出来。muduo 删繁就简，举重若轻; 减少选择余地，生活更简单。\n",
    "\n",
    "`MutexLock` 没有提供`trylock()`函数，因为我没有在生成代码中用过它。我想不出什么时候程序需要\"试着去锁一锁\"，或许我写过的代码太简单了\n",
    "> `trylock` 的一个用途是用来观察lock contention，见[RWC]\"Consider using nonblocking synchronization routines to monitor contention.\"\n",
    "\n",
    "`Condition class` 的实现有点意思。`Pthreads condition variable` 允许在 `wait()`的时候指定 mutex，但是我想不出有什么理由一个 condition variable会和不同的mutex配合使用。Java 的intrinsic condition 和Condition class都不支持这么做，因此我觉得可以放弃这一灵活性，老老实实地一对一好了。\n",
    "\n",
    "相反，`boost::thread`的 condition_variable是在 wait的时候指定mutex，请参观其同步原语的庞杂设计∶\n",
    "- Concept有四种 `Lockable、TimedLockable、SharedLockable、UpgradeLockable`。\n",
    "- Lock 有六种∶`lock_guard、unique_lock、shared_lock、upgrade_lock、upgrade_to_unique_lock、scoped_try_lock`。\n",
    "- Mutex 有七种∶`mutex、try_mutex、timed_mutex、recursive_mutex、recursive_try_mutex、recursive_timed_mutex、shared_mutex`。\n",
    "\n",
    "恕我愚钝，见到 `boost::thread` 这样如 Rube Goldberg Machine一样让人眼花缭乱的库，我只得三揖绕道而行。很不幸C++11 的线程库也采纳了这套方案。这些class 名字也很无厘头，为什么不老老实实用 `readers_writer_lock `这样的通俗名字呢?非得增加精神负担，自己发明新名字。我不愿为这样的灵活性付出代价，宁愿自己做几个简简单单的一看就明白的 class 来用，这种简单的几行代码的\"轮子\"造造也无妨。提供灵活性固然是本事，然而在不需要灵活性的地方把代码写死，更需要大智慧。\n",
    "\n",
    "下面这个`muduo::Condition class`简单地封装了`Pthreads condition variable`，用起来也容易，见本节前面的例子。这里我用`notify/notifyAll`作为函数名，因为signal有别的含义，C++里的 signal/slot、C里的 signal handler等等。就别 overload这个术语了。\n",
    "\n",
    "- [Condition class](../../recipes/thread/Condition.h)\n",
    "\n",
    "如果一个 class要包含`MutexLock` 和 `Condition`，请注意它们的声明顺序和初始化顺序，`mutex_`应先于`condition_`构造，并作为后者的构造参数∶\n",
    "\n",
    "- [class CountDownLatch](../../recipes/thread/CountDownLatch.h)\n",
    "\n",
    "请允许我再次强调，虽然本章花了大量篇幅介绍如何正确使用mutex 和 condition variable，但并不代表我鼓励到处使用它们。这两者都是非常底层的同步原语，主要用来实现更高级的并发编程工具。一个多线程程序里如果大量使用mutex 和condition variable来同步，基本跟用铅笔刀锯大树（孟岩语）没啥区别。\n",
    "\n",
    "在程序里使用`Pthreads`库有一个额外的好处∶分析工具认得它们，懂得其语意。线程分析工具如 Intel Thread Checker和 Valgrind-Helgrind 等能识别`Pthreads` 调用，并依据 happens-before关系分析程序有无data race。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.5 线程安全的 Singleton 实现\n",
    "\n",
    "研究 Singleton 的线程安全实现的历史会发现很多有意思的事情，人们一度认为double checked locking（缩写为 DCL）是王道，兼顾了效率与正确性。后来有\"神牛\"指出由于乱序执行的影响，DCL是靠不住的。Java开发者还算幸运，可以借助内部静态类的装载来实现。C++就比较惨，要么次次锁，要么 eager initialize，或者动用memory barrier这样的\"大杀器\"。接下来 Java 5修订了内存模型，并给 volatile赋予了acquire/release 语义，这下 DCL（with volatile）又是安全的了。然而 C++的内存模型还在修订中，C++的 volatile 目前还不能（将来也难说）保证 DCL的正确性（只在Visual C++2005及以上版本有效）。\n",
    "> [C++11制定的内存模型](http://scottmeyers.blogspot.com/2012/04/information-on-c11-memory-model.html) \n",
    "\n",
    "其实没那么麻烦，在实践中用 `pthread_once` 就行\n",
    "- [class Singleton](../../recipes/thread/Singleton.h)\n",
    "\n",
    "上面这个Singleton 没有任何花哨的技巧，它用`pthread_once_t`来保证 lazy-initialization 的线程安全。线程安全性由 `Pthreads` 库保证，如果系统的`Pthreads`库有bug，那就认命吧，多线程程序反正也不可能正确执行了。\n",
    "\n",
    "使用方法也很简单∶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "Foo& foo = Singleton<Foo>::instance();"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这个 `Singleton` 没有考虑对象的销毁。在长时间运行的服务器程序里，这不是一个问题，反正进程也不打算正常退出（S9.2.2）。在短期运行的程序中，程序退出的时候自然就释放所有资源了（前提是程序里不使用不能由操作系统自动关闭的资源，比如跨进程的 mutex）。在实际的 `muduo::Singleton class`中，通过 `atexit(3)`提供了销毁功能，聊胜于无罢了。\n",
    "\n",
    "另外，这个`Singleton` 只能调用默认构造函数，如果用户想要指定`T`的构造方式，我们可以用模板特化（template specialization）技术来提供一个定制点，这需要引入另一层间接（another level of indirection）。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "| 方法                   | 标准依赖                          | 可移植性                  | 实现复杂度 | 典型场景                   | 资源清理控制       |\n",
    "| ---------------------- | --------------------------------- | ------------------------- | ---------- | -------------------------- | ------------------ |\n",
    "| 函数内静态局部变量     | C++11 及以上                      | 高                        | 低         | 通用、推荐默认方案         | 受静态析构顺序影响 |\n",
    "| `std::call_once`       | C++11 及以上                      | 高                        | 中         | 需要自定义清理或延迟初始化 | 可通过智能指针控制 |\n",
    "| `pthread_once`         | C++98+（POSIX 环境）              | 仅 POSIX                  | 中         | 需兼容旧编译器/平台        | 自行管理           |\n",
    "| Double-Checked Locking | C++11+（必须用 `std::atomic` 等） | 中（依赖 C++11 内存模型） | 高         | 对性能极端敏感场景（罕见） | 自行管理           |"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.6 `sleep(3)`不是同步原语\n",
    "\n",
    "我认为 `sleep()/usleep()/nanosleep()` 只能出现在测试代码中，比如写单元测试的时候; 或者用于有意延长临界区，加速复现死锁的情况，就像S2.1.2示范的那样。`sleep` 不具备memory barrier语义，它不能保证内存的可见性，见 p.84的例子。\n",
    "> 涉及时间的单元测试不那么好写，短的如一两秒，可以用 `sleep();`长的如一小时、一天，则得想其他办法，比如把算法提取出来并把时间注人进去。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "bool running = false; // 全局标志\n",
    "\n",
    "void threadFunc()\n",
    "{\n",
    "    while(running)\n",
    "    {\n",
    "        // get task from queue\n",
    "    }\n",
    "}\n",
    "\n",
    "void start()\n",
    "{\n",
    "    muduo::Thread t(threadFunc);\n",
    "    t.start();\n",
    "    running = true; // 应该放到t.start()之前\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "生产代码中线程的等待可分为两种：\n",
    "\n",
    "- 一种是等待资源可用（要么等在`select/poll/epoll_wait`上，要么等在条件变量上）；\n",
    "> 等待`BlockingQueue/CountDownLatch` 亦可归人此类。\n",
    "- 一种是等着进入临界区（等在mutex上）以便读写共享数据。\n",
    "\n",
    "后一种等待通常极短，否则程序性能和伸缩性就会有问题。\n",
    "\n",
    "在程序的正常执行中，如果需要等待一段已知的时间，应该往 `event loop`里注册一个 `timer`，然后在 `timer`的回调函数里接着干活，因为线程是个珍贵的共享资源，不能轻易浪费（阻塞也是浪费）。如果等待某个事件发生，那么应该采用条件变量或IO事件回调，不能用`sleep`来轮询。\n",
    "\n",
    "不要使用下面这种业余做法∶"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "while(true){\n",
    "    if(!dataAvailable)\n",
    "        sleep(some_time)\n",
    "    else\n",
    "        consumeData();\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "如果多线程的安全性和效率要靠代码主动调用`sleep`来保证，这显然是设计出了问题。等待某个事件发生，正确的做法是用`select()`等价物或Condition，抑或（更理想地）高层同步工具；在用户态做轮询(polling)是低效的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 2.7 归纳与总结\n",
    "\n",
    "前面几节内容归纳如下∶\n",
    "\n",
    "- 线程同步的四项原则，尽量用高层同步设施（线程池、队列、倒计时）;\n",
    "- 使用普通互斥器和条件变量完成剩余的同步任务，采用RAII惯用手法(idiom)和Scoped Locking。\n",
    "\n",
    "用好这几样东西，基本上就能应付多线程服务端开发的各种场合。或许有人会觉得性能没有发挥到极致。我认为，应该先把程序写正确（并尽量保持清晰和简单），然后再考虑性能优化，如果确实还有必要优化的话。这在多线程下仍然成立。让一个正确的程序变快，远比“让一个快的程序变正确”容易得多。\n",
    "\n",
    "在现代的多核计算背景下，多线程是不可避免的。尽管在一定程度上可以通过framework来屏蔽，让你感觉像是在写单线程程序，比如Java Servlet。了解under the hood发生了什么对于编写这种程序也会有帮助。\n",
    "\n",
    "多线程编程是一项重要的个人技能，不能因为它难就本能地排斥，现在的软件开发比起10年、20年前已经难了不知道多少倍。掌握多线程编程，才能更理智地选择用还是不用多线程，因为你能预估多线程实现的难度与收益，在一开始做出正确的选择。要知道把一个单线程程序改成多线程的，往往比从头实现一个多线程的程序更困难。要明白多线程编程中哪些是能做的，哪里是一般程序员应该避开的雷区。\n",
    "\n",
    "掌握同步原语和它们的适用场合是多线程编程的基本功。以我的经验，熟练使用文中提到的同步原语，就能比较容易地编写线程安全的程序。本文没有考虑signal对多线程编程的影响(§ 4.10)，Unix的signal在多线程下的行为比较复杂，一般要靠底层的网络库（如Reactor）加以屏蔽，避免干扰上层应用程序的开发。\n",
    "\n",
    "通篇来看，“效率”并不是我的主要考虑点，我提倡正确加锁而不是自己编写lock-free算法（使用原子整数除外），更不要想当然地自己发明[同步设施](https://www.usenix.org/events/osdi10/tech/full_papers/Xiong.pdf)。在没有实测数据支持的情况下，妄谈哪种做法效率更高是[靠不住的](http://pdos.csail.mit.edu/papers/linux:osdi10.pdf)，不能听信传言或凭感觉“优化”。很多人误认为用锁会让程序变慢，其实真正影响性能的不是锁，而是锁争用[(lock contention)](http://preshing.com/20111118/locks-arent-slow-lock-contention-is)。在程序的复杂度和性能之前取得平衡，并考虑未来两三年扩容的可能（无论是CPU变快、核数变多，还是机器数量增加、网络升级）。我认为在分布式系统中，多机伸缩性(scale out)比单机的性能优化更值得投入精力。\n",
    "\n",
    "本章内容记录了我目前对多线程编程的理解，用文中介绍的手法，我能化繁为简，编写容易验证其正确性的多线程程序，解决自己面临的全部多线程编程任务。如果本章的观点与你的经验不合，比如你使用了我没有推荐使用的技术或手法（共享内存、信号量等等），只要你理由充分，但行无妨。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# * 2.8 借 shared_ptr 实现 copy-on-write\n",
    "\n",
    "本节解决§ 2.1的几个未决问题：\n",
    "\n",
    "- § 2.1.1 `post()`和`traverse()`死锁。\n",
    "- § 2.1.2 把`Request::print()`移出I`nventory::printAll()`临界区。\n",
    "- § 2.1.2 解决`Request`对象析构的race condition。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "本节要解决 §2.1 提出的三个多线程并发问题，它们都源于对共享数据的同步策略不当：\n",
    "1. **§2.1.1 `post()` 和 `traverse()` 之间的死锁**:<br>\n",
    "    原始代码中，`traverse()` 在持有 `mutex` 的情况下遍历 `Foo` 列表，遍历过程中调用 `Foo::doit()`，而 `Foo::doit()` 又会调用 `post()` 再次试图加锁，从而触发“同一线程重复加普通互斥锁”的死锁。\n",
    "2. **§2.1.2 将 `Request::print()` 移出 `Inventory::printAll()` 的临界区**:<br>\n",
    "    直接在临界区内调用 `Request::print()` 会导致锁的持有时间过长，甚至产生跨对象死锁（`Request::print()` 内部还会尝试锁自身的 `mutex_`）。\n",
    "3. **§2.1.2 解决 `Request` 对象析构时的竞态**:<br>\n",
    "    当 `Inventory` 的 `set` 中持有未失效的 `Request*` 时，请求对象析构如果没有同步保护，就有可能在析构期间被其他线程访问，引发段错误或未定义行为。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "然后再示范用普通 mutex 替换读写锁。解决办法都基于同一个思路，那就是用`shared_ptr` 来管理共享数据。\n",
    "\n",
    "原理如下∶\n",
    "\n",
    "- `shared_ptr`是引用计数型智能指针，如果当前只有一个观察者，那么引用计数的值为`1`。\n",
    "> 在实际代码中判断`shared_ptr::unique()`是否为 `true`\n",
    "- 对于write端，如果发现引用计数为`1`，这时可以安全地修改共享对象，不必担心有人正在读它。\n",
    "- 对于read端，在读之前把引用计数加`1`，读完之后减`1`，这样保证在读的期间其引用计数大于`1`，可以阻止并发写。\n",
    "- 比较难的是，对于write端，如果发现引用计数大于`1`，该如何处理？`sleep()`一小段时间肯定是错的。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "**核心思路：用 `shared_ptr` 驱动 Copy-on-Write**\n",
    "\n",
    "`shared_ptr` 的引用计数让我们可以轻松判断**共享对象当前是否只有一个访问者**。当 `shared_ptr::unique()` 返回 `true` 时，意味着当前只有调用者持有该对象，安全进行原地修改 (in-place)。如果 `unique()` 返回 `false`，说明还有其他观察者（通常是读线程）在共享这个对象，此时就 **复制一份副本再修改**。这就是 Copy-on-Write (COW) 的核心。\n",
    "\n",
    "**原理概述**\n",
    "\n",
    "- 写路径\n",
    "```cpp\n",
    "if (ptr.unique()) {\n",
    "    // 原地修改\n",
    "} else {\n",
    "    // 复制副本 -> 修改副本 -> 替换原始 shared_ptr\n",
    "}\n",
    "```\n",
    "- 读路径: 在读取之前获取一个 `shared_ptr` 副本（通常是局部变量），这样引用计数 +1，保证读操作期间不会因为写线程的替换而破坏数据。\n",
    "\n",
    "关键是整套机制要求：**只有“检测 + 决策 + 修改 + 替换”这一串操作完全在临界区内**，否则就会出现竞态或再次死锁。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "先来看一个简单的例子，解决§ 2.1.1中的`post()`和`traverse()`死锁。数据结构改成：\n",
    "- [CopyOnWrite_test.cc](../../recipes/thread/test/CopyOnWrite_test.cc#L15)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "typedef std::vector<Foo> FooList;\n",
    "typedef boost::shared_ptr<FooList> FooListPtr;\n",
    "FooListPtr g_foos;\n",
    "MutexLock mutex;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "在read端，用一个栈上局部`FooListPtr`变量当做\"观察者\"，它使得`g_foos`的[引用计数增加](../../recipes/thread/test/CopyOnWrite_test.cc#L38)。`traverse()`函数的临界区是[L36~L40](../../recipes/thread/test/CopyOnWrite_test.cc#L36-L40)，临界区内只读了一次共享变量`g_foos`(这里多线程并发读写`shared_ptr`，因此必须用mutex 保护)，比原来的写法大为缩短。而且多个线程同时调用`traverse()`也不会相互阻塞。\n",
    "\n",
    "- [CopyOnWrite_test.cc](../../recipes/thread/test/CopyOnWrite_test.cc#L33)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void traverse()\n",
    "{\n",
    "  FooListPtr foos; // 副本\n",
    "  {\n",
    "    MutexLockGuard lock(mutex); // 关键：仅保护对 g_foos 的读取\n",
    "    foos = g_foos; // 引用计数 +1\n",
    "    assert(!g_foos.unique());\n",
    "  } // 立刻释放锁\n",
    "\n",
    "  // assert(!foos.unique()); this may not hold\n",
    "\n",
    "  for (std::vector<Foo>::const_iterator it = foos->begin();\n",
    "      it != foos->end(); ++it)\n",
    "  {\n",
    "    it->doit();\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- 临界区极短，只负责取出共享指针。\n",
    "- 读操作持有 `FooListPtr` 副本，确保遍历期间底层数据不会被销毁或突变。\n",
    "- 多线程同时遍历互不阻塞。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "关键看write端的`post()`该如何写。按照前面的描述，如果`g_foos.unique()`为`true`，我们可以放心地在原地(in-place)修改`FooList`。如果`g_foos.unique()`为`false`，说明这时别的线程正在读取`FooList`，我们不能原地修改，而是[复制一份](../../recipes/thread/test/CopyOnWrite_test.cc#L26)，在[副本上修改](../../recipes/thread/test/CopyOnWrite_test.cc#L30)。这样就避免了死锁。\n",
    "\n",
    "- [CopyOnWrite_test.cc](../../recipes/thread/test/CopyOnWrite_test.cc#L20)\n",
    "\n",
    "注意这里[临界区包括整个函数](../../recipes/thread/test/CopyOnWrite_test.cc#L23-L30)，其他写法都是错的。读者可以试着运行这个程序，看看什么时候会[打印](../../recipes/thread/test/CopyOnWrite_test.cc#L27)的消息。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void post(const Foo& f)\n",
    "{\n",
    "  printf(\"post\\n\");\n",
    "  MutexLockGuard lock(mutex); // 整个函数都必须在锁内\n",
    "  if (!g_foos.unique())\n",
    "  {\n",
    "    g_foos.reset(new FooList(*g_foos)); // 复制副本\n",
    "    printf(\"copy the whole list\\n\"); // 练习: 将这句话移出临界区\n",
    "  }\n",
    "  assert(g_foos.unique());\n",
    "  g_foos->push_back(f); // 安全原地修改\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- 先持锁，判断 `g_foos.unique()`。\n",
    "- 如果有人在读，复制一份副本在副本上修改。\n",
    "- 复制完成后，才原地写入，避免了读写竞争和死锁。\n",
    "\n",
    "为什么这套方案正确？\n",
    "\n",
    "1. **读者持有副本**（即 local `FooListPtr foos`），即便写者替换了 `g_foos`，读者仍然访问原有的数据副本，直到副本引用计数归零才会释放。\n",
    "2. **写者若发现有其他读者**，会复制副本进行修改，避免在读者视角数据“凭空”变化。\n",
    "3. **“判定 unique -> 拷贝 -> 修改” 全程受锁保护**，避免了竞态。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "练习：找出以下几种写法的错误。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 错误一: 直接修改g_foos所指的FooList\n",
    "\n",
    "void post(const Foo& f)\n",
    "{\n",
    "    MutexLockGuard lock(mutex);\n",
    "    g_foos->push_back(f); // 若 g_foos 共享给读者，直接修改造成数据撕裂\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- **违反 COW 协议**：COW 的核心是“写操作前必须确保对象独占”。`g_foos` 是一个 `shared_ptr`，如果当前引用计数大于 1，说明还有其他线程持有同一底层容器（`FooList`）。在未复制的情况下直接 `push_back`，会让其他读者瞬间看到半成品或不一致的数据。\n",
    "- **数据撕裂（Torn Read）**：在多线程环境里，读线程可能正在遍历旧版本的 `FooList`。写线程直接在原地插入元素，读线程就可能读到未完全写好的对象或跳过新元素，导致逻辑错误甚至崩溃。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 在持锁的情况下检查 `g_foos.unique()`，如果不是独占所有权，需要复制：\n",
    "\n",
    "MutexLockGuard lock(mutex);\n",
    "if (!g_foos.unique()) {\n",
    "    g_foos.reset(new FooList(*g_foos));  // 复制出独占副本\n",
    "}\n",
    "g_foos->push_back(f);  // 现在安全"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 错误二∶ 试图缩小临界区，把 copying 移出临界区\n",
    "void post(const Foo& f)\n",
    "{\n",
    "    FooListPtr newFoos(new Foolist(*g_foos)); // 未加锁就访问 g_foos -> 数据竞争\n",
    "    newFoos->push_back(f);\n",
    "    MutexLockGuard lock(mutex);\n",
    "    g_foos = newFoos; // 或者 g_foos.swap(newFoos); // 丢失其他同时写入的修改\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- **数据竞争**：第一行在未加锁的情况下解引用了 `g_foos`。如果此时有其他线程在更新 `g_foos`（包括重置指针、修改内部结构），该访问就属于未同步的并发读写，违反 C++ 内存模型，结果未定义。\n",
    "- **丢失写入**：假设在执行 `new FooList(*g_foos)` 到拿锁之间，另一个线程已经成功拷贝并写入了新的数据并赋值回 `g_foos`。当该线程拿到锁后直接 `g_foos = newFoos;`，等于把别人刚写好的内容覆盖掉，导致更新丢失。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 读取共享指针时也需要持有锁，确保快照一致。\n",
    "// 拷贝操作可以在锁内完成；如果担心复制代价大，可先在锁内抓取快照，再在锁外复制底层数据（但要注意与错误三的区别）\n",
    "\n",
    "FooListPtr newFoos;\n",
    "{\n",
    "    MutexLockGuard lock(mutex);\n",
    "    if (!g_foos.unique()) {\n",
    "        g_foos.reset(new FooList(*g_foos)); // 在锁内完成写前复制\n",
    "    }\n",
    "    newFoos = g_foos;  // 如果需要在锁外继续加工，可抓取快照\n",
    "}\n",
    "// 锁外操作 newFoos 的只读部分（若需要写入，应回到锁内并验证版本）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 错误三: 临界区拆成两个小的，把 copying 放到临界区之外\n",
    "void post(const Foo& f)\n",
    "{\n",
    "    FooListPtr oldFoos;\n",
    "    {\n",
    "        MutexLockGuard lock(mutex); \n",
    "        oldFoos = g_foos; // 离开锁后 oldFoos 是快照; 期间可能有新的写入\n",
    "    }\n",
    "\n",
    "    FooListPtr newFoos(new Foolist(*oldFoos));\n",
    "    newFoos->push_back(f);\n",
    "    MutexLockGuard lock(mutex);\n",
    "    g_foos = newFoos; // 或者 g_foos.swap(newFoos);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "- **快照过期**：第一个锁区间结束后，你手里的 `oldFoos` 只是当时的快照。第二个锁区间开始前，其他线程完全可能已经更新了 `g_foos`。\n",
    "- **更新覆盖**：当你在第二个锁区间直接 `g_foos = newFoos` 时，等于无条件用旧快照的修改覆盖当前最新状态，造成其他写线程的更新丢失。\n",
    "\n",
    "这是典型的“读-改-写”操作未保持原子性导致的错误。COW 的更新必须在持锁期间完成完整的流程，否则就会引入 TOCTOU（Time-Of-Check to Time-Of-Use）漏洞。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// 保持整个“检查 → 复制 → 修改 → 赋值”流程处于同一个临界区内，或设计版本检查机制避免覆盖：\n",
    "\n",
    "MutexLockGuard lock(mutex);\n",
    "FooListPtr newFoos = g_foos;\n",
    "if (!newFoos.unique()) {\n",
    "    newFoos.reset(new FooList(*newFoos));\n",
    "}\n",
    "newFoos->push_back(f);\n",
    "g_foos = newFoos;  // 同一个锁区间内完成"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "**所有关于 `g_foos` 的修改和“是否复制”的决策必须在一次加锁期间完成**。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "希望读者先吃透上面举的这个例子，再来看如何用相同的思路解决剩下的问题。\n",
    "\n",
    "解决§ 2.1.2把`Request::print()`移出`Inventory::printAll()`临界区有两个做法。\n",
    "\n",
    "其一很简单，把`requests_`复制一份，在临界区之外遍历这个副本。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void Inventory::printAll() const\n",
    "{\n",
    "    std::set<Request*> requests\n",
    "    {\n",
    "        muduo::MutexLockGuard lock(mutex_); \n",
    "    \trequests = requests_;\n",
    "    }\n",
    "    \n",
    "\t// 遍历局部变量 requests，调用 Request∶∶print()\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这么做有一个明显的缺点，它复制了整个`std::set`中的每个元素，开销可能会比较大。如果遍历期间没有其他人修改`requests_`，那么我们可以减小开销，这就引出了第二种做法。\n",
    "\n",
    "第二种做法的要点是用`shared_ptr`管理`std::set`，在遍历的时候先增加引用计数，阻止并发修改。当然`Inventory::add()`和`Inventory::remove()`也要相应修改，采用本节前面`post()`和`traverse()`的方案。\n",
    "\n",
    "注意目前的方案仍然没有解决`Request`对象析构的race condition，这点还是留作练习吧。\n",
    "\n",
    "一种可能的答案见\n",
    "- [RequestInventory_test2.cc](../../recipes/thread/test/RequestInventory_test2.cc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "> AI:\n",
    "\n",
    "**方法一：复制原始容器**\n",
    "\n",
    "在 `MutualDeadLock.cc` 注释掉的版本里，`printAll()` 先复制 `requests_`："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void Inventory::printAll() const\n",
    "{\n",
    "  std::set<Request*> requests;\n",
    "  {\n",
    "    muduo::MutexLockGuard lock(mutex_);\n",
    "    requests = requests_;    // 拷贝整张表\n",
    "  }\n",
    "\n",
    "  for (auto it = requests.begin(); it != requests.end(); ++it)\n",
    "  {\n",
    "    (*it)->print();          // 临界区外调用，避免交叉锁\n",
    "  }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 简单直接，但每次都复制整个 `std::set`，代价高。\n",
    "- 适用于数据量小或访问不频繁的场景。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**方法二：`shared_ptr` 管理 `std::set`**\n",
    "\n",
    "`RequestInventory_test2.cc` 展示了高效方案："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "typedef std::set<RequestPtr> RequestList;\n",
    "typedef boost::shared_ptr<RequestList> RequestListPtr;\n",
    "\n",
    "RequestListPtr requests_;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "-  读路径（`Inventory::printAll()`）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "RequestListPtr requests = getData();  // getData() 内部短暂持锁\n",
    "for (auto& req : *requests)\n",
    "{\n",
    "  req->print();                       // 临界区外调用，避免锁顺序反转\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "- 写路径（`add` / `remove`）"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "// - 与 `post()` 一样，当 `requests_` 引用计数不唯一时，复制副本后再修改。\n",
    "// - `RequestPtr` 使用 `shared_from_this()`，保证临时快照里持有 `Request` 对象的强引用，防止对象析构被提前触发。\n",
    "\n",
    "void Inventory::add(const RequestPtr& req)\n",
    "{\n",
    "  muduo::MutexLockGuard lock(mutex_);\n",
    "  if (!requests_.unique())\n",
    "  {\n",
    "    requests_.reset(new RequestList(*requests_));\n",
    "    printf(\"Inventory::add() copy the whole list\\n\");\n",
    "  }\n",
    "  requests_->insert(req);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**析构竞态如何处理？**\n",
    "\n",
    "`Request::cancel()` 中调用 `shared_from_this()`："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "void Request::cancel()\n",
    "{\n",
    "  muduo::MutexLockGuard lock(mutex_);\n",
    "  // ...\n",
    "  g_inventory.remove(shared_from_this());\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "这样即使 `Request` 的外部强引用释放，只要 `requests_` 中还持有 `shared_ptr`，对象就不会被销毁。\n",
    "\n",
    "**如果要彻底解决析构竞态**（例如同时存在显式 `delete` 或 `unique_ptr` 场景），需要统一改用 `shared_ptr` 管理 `Request`，并确保析构前不会再被访问。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 用普通mutex替换读写锁的一个例子\n",
    "\n",
    "场景：一个多线程的C++程序，24h x 5.5d运行。有几个工作线程`Thread-Worker{0，1，2，3}`，处理客户发过来的交易请求；另外有一个背景线程ThreadBack-ground，不定期更新程序内部的参考数据。\n",
    "\n",
    "这些线程都跟一个hash表打交道，工作线程只读，背景线程读写，必然要用到一些同步机制，防止数据损坏。\n",
    "\n",
    "这里的示例代码用`std::map`代替hash表，意思是一样的："
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "using namespace std;\n",
    "\n",
    "typedef map<string, vector<pair<string, int>>> Map;"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "`Map`的`key`是用户名，`value`是一个`vector`，里边存的是不同`stock`的最小交易间隔，`vector`已经排好序，可以用二分查找。\n",
    "\n",
    "我们的系统要求工作线程的延迟尽可能小，可以容忍背景线程的延迟略大。一天之内，背景线程对数据更新的次数屈指可数，最多一小时一次，更新的数据来自于网络，所以对更新的及时性不敏感。`Map`的数据量也不大，大约一千多条数据。\n",
    "\n",
    "最简单的同步办法是用读写锁：工作线程加读锁，背景线程加写锁。但是读写锁的开销比普通mutex要大，而且是写锁优先，会阻塞后面的读锁。如果工作线程能用最普通的非重入mutex实现同步，就不必用读写锁，这能降低工作线程延迟。我们借助`shared_ptr`做到了这一点：\n",
    "\n",
    "- [class CustomerData](../../recipes/thread/test/Customer.cc)\n",
    "\n",
    "`CustomerData::query()`就用前面说的引用计数加`1`的办法，用局部`MapPtr data`变量来持有`Map`，防止并发修改。\n",
    "\n",
    "- [CustomerData::query](../../recipes/thread/test/Customer.cc#L41)\n",
    "\n",
    "关键看`CustomerData::update()`怎么写。既然要更新数据，那肯定得加锁，如果这时候其他线程正在读，那么不能在原来的数据上修改，得创建一个副本，在副本上修改，修改完了再替换。如果没有用户在读，那么就能直接修改，节约一次Map拷贝。\n",
    "\n",
    "- [CustomerData::update()](../../recipes/thread/test/Customer.cc#L52)\n",
    "\n",
    "注意其中用了`shared_ptr::unique()`来判断是不是有人在读，如果有人在读，那么我们不能直接修改，因为`query()`并没有全程加锁，只在`getData()`内部有锁。`shared_ptr::swap()`把`data_`替换为新副本，而且我们还在锁里，不会有别的线程来读，可以放心地更新。如果别的reader线程已经刚刚通过`getData()`拿到了`MapPtr`，它会读到稍旧的数据。这不是问题，因为数据更新来自网络，如果网络稍有延迟，反正reader线程也会读到旧的数据\n",
    "\n",
    "如果每次都更新全部数据，而且始终是在同一个线程更新数据，临界区还可以进一步缩小。"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "vscode": {
     "languageId": "cpp"
    }
   },
   "outputs": [],
   "source": [
    "MapPtr parseData(const string& message); // 解析收到的消息，返回新的 MapPtr\n",
    "\n",
    "void CustomerData::update(const string& message) // 函数原型有变，此时网络上传来的是完整的 Map 数据\n",
    "{\n",
    "    MapPtr newData = parseData(message); // 解析新数据，在临界区之外\n",
    "    if (newData)\n",
    "    {\n",
    "        MutexLockGuard lock(mutex_); \n",
    "        data_.Swap(newData); // 不要用 data_= newData;\n",
    "\n",
    "    } // 旧数据的析构也在临界区外，进一步缩短了临界区\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "据我们测试，大多数情况下更新都是在原来数据上进行的，拷贝的比例还不到1%很高效。更准确地说，这不是copy-on-write，而是 copy-on-other-reading。\n",
    "\n",
    "我们将来可能会采用无锁数据结构，不过目前这个实现已经非常好，可以满足我们的要求。\n",
    "\n",
    "本节介绍的做法与read-copy-update颇有相似之处，但理解起来要容易得多。"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "C++17",
   "language": "C++17",
   "name": "xcpp17"
  },
  "language_info": {
   "codemirror_mode": "text/x-c++src",
   "file_extension": ".cpp",
   "mimetype": "text/x-c++src",
   "name": "C++17",
   "version": "17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
